package com.base.test.rabbitmq.service;

import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.rabbitmq.client.Channel;
import com.base.lock.executor.RedisTemplateLockExecutor;
import com.base.test.rabbitmq.constant.Constant;
import com.base.test.rabbitmq.entity.MqReceiveEntity;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.beans.factory.annotation.Autowired;
import java.util.UUID;
import static com.base.test.rabbitmq.constant.Constant.LOCK_SUCCESS;
import static com.base.test.rabbitmq.constant.Constant.ONE_MINUTE;

/**
 * 所有消费者父类
 * @author qiaodong
 * @since 2020-10-12
 */
@Slf4j
public abstract class AbstractConsumerReceiver {

    @Autowired
    private MqReceiveService mqReceiveService;

    @Autowired
    private RedisTemplateLockExecutor redisTemplateLockExecutor;

    public void receive(String queue,String appCode,String bussinessType,
                           Message message,Channel channel)  throws Exception {
        String uuid = UUID.randomUUID().toString();
        String json = new String(message.getBody());
        log.info("接收应用["+appCode+"]业务场景["+bussinessType+"]推送信息:{}",json);
        JSONObject jsonObj = JSONObject.parseObject(json);
        String msgId = String.valueOf(jsonObj.get("msg_id"));
        String redisKey = String.format(Constant.MQ_LOCK, msgId, appCode);
        try {
            int result = Constant.ZERO;
            //重试场景先查看DB是否保存了已发送消息
            String lockFlag = redisTemplateLockExecutor.acquire(redisKey,uuid,ONE_MINUTE,ONE_MINUTE);
            if (!LOCK_SUCCESS.equals(lockFlag)) {
                log.info("消息id[{}]消费应用[{}]业务场景[{}]消息重复",msgId, appCode, bussinessType);
            }else{
                MqReceiveEntity mqReceive = mqReceiveService.getByAppCodeMsgId(appCode,msgId);
                if(null==mqReceive){
                    MqReceiveEntity msg = makeMsg(msgId,queue,json,appCode,bussinessType,
                            Constant.ONE,Constant.ZERO,"接收消息入库");
                    result = mqReceiveService.checkReceiveInsertOrUpdate(msg);
                    //接收消息插入或更新消费表成功后再消费消息
                    if(result > Constant.ZERO){
                        realityReceive(json);
                        //消费成功更新DB
                        MqReceiveEntity msg1 = makeMsg(msgId,queue,json,appCode,bussinessType,
                                Constant.THREE,Constant.ZERO,"消费消息成功");
                        mqReceiveService.checkReceiveInsertOrUpdate(msg1);
                    }
                }else{
                    realityReceive(json);
                    //乐观锁方式进行重试次数累加
                    mqReceiveService.addReceiveRetryCount(appCode,msgId,mqReceive.getRetryCount(),
                            Constant.THREE,"消费消息成功");
                }
                log.info("消费应用[{}]业务场景[{}]消息成功",appCode,bussinessType);
            }
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }catch (Exception e){
            log.error("消费应用[{}]消息异常，业务场景[{}],消息内容:{},错误信息:{}",appCode,bussinessType,json,e);
            MqReceiveEntity mqReceive = mqReceiveService.getByAppCodeMsgId(appCode,msgId);
            //消费失败更新DB
            if(null==mqReceive){
                MqReceiveEntity msg = makeMsg(msgId,queue,json,appCode,bussinessType,
                        Constant.TWO,Constant.ZERO,"消费消息失败");
                mqReceiveService.checkReceiveInsertOrUpdate(msg);
            }else{
                //消费失败消息乐观锁方式进行重试次数累加
                mqReceiveService.addReceiveRetryCount(appCode,msgId,mqReceive.getRetryCount(),
                        Constant.TWO,"消费失败重试");
            }
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false,false);
            throw e;
        }finally{
            channel.close();
            redisTemplateLockExecutor.releaseLock(redisKey,uuid,null);
        }
    }

    /**
     * 实际消费消息方法
     * @param json
     */
    public abstract void realityReceive(String json) throws Exception;

    /**
     * 构建消息内容
     * @param msgId
     * @param queue
     * @param json
     * @param appCode
     * @param bussinessType
     * @param status
     * @param retryCount
     * @param remark
     * @return
     */
    private MqReceiveEntity makeMsg(String msgId,String queue,String json,String appCode,
                                    String bussinessType,Integer status,Integer retryCount,
                                    String remark){
        MqReceiveEntity mqSend = new MqReceiveEntity();
        mqSend.setId(IdWorker.getId());
        mqSend.setMsgId(msgId);
        mqSend.setQueue(queue);
        mqSend.setMsgJson(json);
        mqSend.setAppCode(appCode);
        mqSend.setBusinessType(bussinessType);
        //状态 1：待定 2：失败 3：成功
        mqSend.setStatus(status);
        if(null!=retryCount){
            mqSend.setRetryCount(retryCount);
        }
        mqSend.setRemark(remark);
        return mqSend;
    }

}
